/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements.  See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership.  The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with the License.  You may obtain
 * a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.apache.zookeeper.server;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.SessionExpiredException;
import org.apache.zookeeper.common.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

/**
 * This is a full featured SessionTracker. It tracks session in grouped by tick
 * interval. It always rounds up the tick interval to provide a sort of grace
 * period. Sessions are thus expired in batches made up of sessions that expire
 * in a given interval.
 */
public class SessionTrackerImpl extends ZooKeeperCriticalThread implements SessionTracker {
    private static final Logger LOG = LoggerFactory.getLogger(SessionTrackerImpl.class);

    /**
     * sessionId -> Session
     */
    HashMap<Long, SessionImpl> sessionsById = new HashMap<Long, SessionImpl>();

    /**
     * sessionId -> SessionSet
     */
    HashMap<Long, SessionSet> sessionSets = new HashMap<Long, SessionSet>();

    /**
     * sessionId -> sessionTimeout
     *
     * 备注：我个人认为这个Map没有实际作用。
     * 查找这个变量的使用，只在{@link SessionTrackerImpl#addSession }方法中put数据，
     * 在{@link SessionTrackerImpl#removeSession}中remove掉，而且还没有使用remove方法的返回值。
     * 所以我认为这个变量没啥实际作用。
     */
    ConcurrentHashMap<Long, Integer> sessionsWithTimeout;

    /**
     * 创建session时，下一个sessionId
     */
    long nextSessionId = 0;

    /**
     * 系统当前时间
     */
    volatile long currentTime;

    /**
     * 超时间隔
     * <pre>
     * 在‘分桶策略’中，ZooKeeper服务器会把时间分成相同长度的时间间隔：expirationInterval，默认为tickTime值（也就是默认2000毫秒）。
     * 为了方便对于多个会话进行超时检查：真实实现的expireTime超时时间点计算方式如下：
     *      expireTime_0 = currentTime + sessionTimeout
     *      expireTime = (expireTime_0 / expirationInterval + 1) * expirationInterval
     * 也就是说横坐标时间轴的expireTime总是expirationInterval的整数倍。
     * 而横坐标时间轴上每个 expireTime时间点，都是‘session超时任务’的执行时间点。
     * </pre>
     */
    int expirationInterval;

    /**
     * 下一个超时时间点
     *
     * 在run()方法中，根据这个时间点来判断当前事件是否能够执行‘过期session清理’任务；
     * 在执行一次任务之后
     */
    long nextExpirationTime;


    /**
     * session超时处理器
     */
    SessionExpirer expirer;

    /**
     * 线程是否运行
     */
    volatile boolean running = true;


    public SessionTrackerImpl(SessionExpirer expirer,
                              ConcurrentHashMap<Long, Integer> sessionsWithTimeout, int tickTime,
                              long sid, ZooKeeperServerListener listener) {
        super("SessionTracker", listener);
        this.expirer = expirer;
        this.expirationInterval = tickTime;
        this.sessionsWithTimeout = sessionsWithTimeout;
        nextExpirationTime = roundToInterval(Time.currentElapsedTime());
        this.nextSessionId = initializeNextSession(sid);
        for (Entry<Long, Integer> e : sessionsWithTimeout.entrySet()) {
            addSession(e.getKey(), e.getValue());
        }
    }

    public static long initializeNextSession(long id) {
        long nextSid = 0;
        nextSid = (Time.currentElapsedTime() << 24) >>> 8;
        nextSid = nextSid | (id << 56);
        return nextSid;
    }

    private long roundToInterval(long expireTime_0) {
        return (expireTime_0 / expirationInterval + 1) * expirationInterval;
    }

    @Override
    synchronized public void dumpSessions(PrintWriter pwriter) {
        pwriter.print("Session Sets (");
        pwriter.print(sessionSets.size());
        pwriter.println("):");
        ArrayList<Long> keys = new ArrayList<Long>(sessionSets.keySet());
        Collections.sort(keys);
        for (long time : keys) {
            pwriter.print(sessionSets.get(time).sessions.size());
            pwriter.print(" expire at ");
            pwriter.print(new Date(time));
            pwriter.println(":");
            for (SessionImpl s : sessionSets.get(time).sessions) {
                pwriter.print("\t0x");
                pwriter.println(Long.toHexString(s.sessionId));
            }
        }
    }

    @Override
    synchronized public String toString() {
        StringWriter sw = new StringWriter();
        PrintWriter pwriter = new PrintWriter(sw);
        dumpSessions(pwriter);
        pwriter.flush();
        pwriter.close();
        return sw.toString();
    }

    @Override
    synchronized public void run() {
        try {
            while (running) {
                currentTime = Time.currentElapsedTime();
                // 1、如果当前时间还没到 下一次超时时间点，则让线程wait 间隔的事件。
                if (nextExpirationTime > currentTime) {
                    this.wait(nextExpirationTime - currentTime);
                    continue;
                }

                // 2、在这里就一定到了 超时时间点。下面就是处理session过期任务：
                SessionSet set;
                // 拿出当前过期时间点的所有session
                set = sessionSets.remove(nextExpirationTime);
                if (set != null) {
                    for (SessionImpl s : set.sessions) {
                        // for循环设置session closing = true
                        setSessionClosing(s.sessionId);
                        // 然后交给session过期处理器处理。
                        expirer.expire(s);
                    }
                }

                // 3、过期任务处理完毕，设置下一次超时任务执行时间点。
                nextExpirationTime += expirationInterval;
            }
        } catch (InterruptedException e) {
            handleException(this.getName(), e);
        }
        LOG.info("SessionTrackerImpl exited loop!");
    }

    @Override
    synchronized public void setSessionClosing(long sessionId) {
        if (LOG.isTraceEnabled()) {
            LOG.info("Session closing: 0x" + Long.toHexString(sessionId));
        }
        SessionImpl s = sessionsById.get(sessionId);
        if (s == null) {
            return;
        }
        s.isClosing = true;
    }

    @Override
    synchronized public boolean touchSession(long sessionId, int timeout) {
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG, ZooTrace.CLIENT_PING_TRACE_MASK, "SessionTrackerImpl --- Touch session: 0x" + Long.toHexString(sessionId) + " with timeout " + timeout);
        }

        SessionImpl session = sessionsById.get(sessionId);
        // Return false, if the session doesn't exists or marked as closing
        if (session == null || session.isClosing()) {
            return false;
        }

        long expireTime_0 = Time.currentElapsedTime() + timeout; // 这是session的实际超时事件
        // 使用分桶策略，计算session超时时间
        long expireTime = roundToInterval(expireTime_0);
        if (session.tickTime >= expireTime) {
            return true;
        }

        // 在这个超时时间点，会有很多session，这里SessionSet用于封装这些session。
        SessionSet set = sessionSets.get(session.tickTime);
        if (set != null) {
            // 从sessionSet移除session。
            set.sessions.remove(session);
        }
        // 为此session设置下一个 超时时间点
        session.tickTime = expireTime;


        // 获取下一个超时时间点的sessionSet，并把session添加到set中。
        set = sessionSets.get(expireTime);
        if (set == null) {
            set = new SessionSet();
            sessionSets.put(expireTime, set);
        }
        set.sessions.add(session);
        return true;
    }

    @Override
    synchronized public void removeSession(long sessionId) {
        SessionImpl s = sessionsById.remove(sessionId);
        sessionsWithTimeout.remove(sessionId);
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
                                     "SessionTrackerImpl --- Removing session 0x" + Long.toHexString(sessionId));
        }
        if (s != null) {
            SessionSet set = sessionSets.get(s.tickTime);
            // Session expiration has been removing the sessions
            if (set != null) {
                set.sessions.remove(s);
            }
        }
    }

    @Override
    public void shutdown() {
        LOG.info("Shutting down");

        running = false;
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(),
                                     "Shutdown SessionTrackerImpl!");
        }
    }

    @Override
    synchronized public long createSession(int sessionTimeout) {
        addSession(nextSessionId, sessionTimeout);
        return nextSessionId++;
    }

    @Override
    synchronized public void addSession(long id, int sessionTimeout) {
        sessionsWithTimeout.put(id, sessionTimeout);
        if (sessionsById.get(id) == null) {
            SessionImpl s = new SessionImpl(id, sessionTimeout, 0);
            sessionsById.put(id, s);
            if (LOG.isTraceEnabled()) {
                ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK, "SessionTrackerImpl --- Adding session 0x" + Long.toHexString(id) + " " + sessionTimeout);
            }
        } else {
            if (LOG.isTraceEnabled()) {
                ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK, "SessionTrackerImpl --- Existing session 0x" + Long.toHexString(id) + " " + sessionTimeout);
            }
        }
        touchSession(id, sessionTimeout);
    }

    @Override
    synchronized public void checkSession(long sessionId, Object owner) throws KeeperException.SessionExpiredException, KeeperException.SessionMovedException {
        SessionImpl session = sessionsById.get(sessionId);
        if (session == null || session.isClosing()) {
            throw new KeeperException.SessionExpiredException();
        }
        if (session.owner == null) {
            session.owner = owner;
        } else if (session.owner != owner) {
            throw new KeeperException.SessionMovedException();
        }
    }

    @Override
    synchronized public void setOwner(long id, Object owner) throws SessionExpiredException {
        SessionImpl session = sessionsById.get(id);
        if (session == null || session.isClosing()) {
            throw new KeeperException.SessionExpiredException();
        }
        session.owner = owner;
    }

    public static class SessionImpl implements Session {
        final long sessionId;
        final int timeout;
        /**
         * session的超时时间点。
         * 这个变量为long类型：由于zk管理session超时时间，使用‘分桶策略’。其值接‘近于’当前事件加上timeout
         */
        long tickTime;
        boolean isClosing;
        Object owner;

        SessionImpl(long sessionId, int timeout, long expireTime) {
            this.sessionId = sessionId;
            this.timeout = timeout;
            this.tickTime = expireTime;
            isClosing = false;
        }

        @Override
        public long getSessionId() {
            return sessionId;
        }

        @Override
        public int getTimeout() {
            return timeout;
        }

        @Override
        public boolean isClosing() {
            return isClosing;
        }
    }

    static class SessionSet {
        HashSet<SessionImpl> sessions = new HashSet<SessionImpl>();
    }
}
